package com.intct.func;

import com.intct.hbase.bean.LocusRedisBean;
import com.intct.util.RedisUtil;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import redis.clients.jedis.Jedis;

/**
 * @author gufg
 * @since 2025-07-04 14:49
 */
public class RedisSinkFunc extends RichSinkFunction<LocusRedisBean> {

    private Jedis jedis;

    @Override
    public void open(Configuration parameters) throws Exception {
        jedis = RedisUtil.getConnection();
    }

    @Override
    public void invoke(LocusRedisBean locusRedis, Context context) throws Exception {
        /*
         *      zset  1）那个字段 做为分数？？？ 字段选择time字段  2）值(经、纬度) lng+","+lat 3) key 订单
         *      set   1) key (当--curr、历--his) 2) 订单号
         *      订单结束："end" time字段=9999999999   经、纬度
         */

        boolean isGameOver = locusRedis.isGameOver();

        // 向有序集合中插入轨迹数据
        jedis.zadd(locusRedis.getOrderId(), locusRedis.getScore(), locusRedis.getValue());

        // 如果订单结束标识==将数据移交到历史订单集合
        if (isGameOver) {
            // 历史订单集合
            jedis.sadd("his", locusRedis.getOrderId());
            // 删除当前集合中的订单
            jedis.srem("curr", locusRedis.getOrderId());
        } else {
            // 如果订单未结束 == 向当前订单集合中增加
            jedis.sadd("curr", locusRedis.getOrderId());
        }

    }

    @Override
    public void close() throws Exception {
        RedisUtil.closeConnection(jedis);
    }
}
